package com.mydemo.likedy.consumer;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mydemo.likedy.Constant.RedisConstant;
import com.mydemo.likedy.config.RabbitMQConfig;
import com.mydemo.likedy.pojo.entity.HotRankBatchMessage;
import com.mydemo.likedy.pojo.entity.HotRankEndMessage;
import com.mydemo.likedy.pojo.entity.video.Video;
import com.mydemo.likedy.pojo.vo.HotVideo;
import com.mydemo.likedy.schedul.TopK;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Comparator;
import java.util.List;
import java.util.PriorityQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class HotRankConsumer {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    // 使用ConcurrentMap存储每个任务的TopK
    private final ConcurrentMap<String, TopK> taskTopKMap = new ConcurrentHashMap<>();
    @Autowired
    private RabbitTemplate rabbitTemplate;

    Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
    ObjectMapper om = new ObjectMapper();

    {
        jackson2JsonRedisSerializer.setObjectMapper(om);
    }
    // 处理批次消息
    @RabbitListener(queues = RabbitMQConfig.HOT_RANK_QUEUE)
    public void processMessage(Message message, Channel channel) throws IOException {
        String routingKey = message.getMessageProperties().getReceivedRoutingKey();
        String taskId = (String) message.getMessageProperties().getHeaders().get("taskId");
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            if (RabbitMQConfig.HOT_RANK_ROUTING_KEY.equals(routingKey)) {
                HotRankBatchMessage batchMsg = (HotRankBatchMessage) rabbitTemplate.getMessageConverter().fromMessage(message);
                processBatch(taskId, batchMsg);
            } else if (RabbitMQConfig.HOT_RANK_END_ROUTING_KEY.equals(routingKey)) {
                HotRankEndMessage endMsg = (HotRankEndMessage) rabbitTemplate.getMessageConverter().fromMessage(message);
                finalizeHotRank(taskId, endMsg);
            }

            // 手动确认消息
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("处理消息失败: taskId={}, routingKey={}", taskId, routingKey, e);

            // 处理失败，拒绝消息（不重新入队）
            // channel.basicReject(deliveryTag, false);

//            //   或者重新入队（最多尝试3次）
//             Integer retryCount = rabbitTemplate.getRetryCount(message);
//             if (retryCount < 3) {
//                 channel.basicReject(deliveryTag, true); // 重新入队
//             } else {
                 channel.basicReject(deliveryTag, false); // 丢弃或进入死信队列
//             }
     }
    }

    private void processBatch(String taskId, HotRankBatchMessage batch) {
        // 初始化或获取TopK (每个任务独立)
        TopK topK = taskTopKMap.computeIfAbsent(taskId, k -> new TopK(10, new PriorityQueue<>(Comparator.comparing(HotVideo::getHot))));

        for (Video video : batch.getVideos()) {
            // 使用统一的baseTime计算
            long t = batch.getBaseTime().getTime() - video.getGmtCreated().getTime();
            double hot = computeHot(video, t);

            topK.add(new HotVideo(hot, video.getId(), video.getTitle()));
        }

        log.debug("处理批次完成: taskId={}, batchIndex={}, 当前TopK大小={}",
                taskId, batch.getBatchIndex(), topK.get().size());
    }

    /**
     * 热度计算
     * @param video
     * @param timeDiffMillis
     * @return
     */

    private double computeHot(Video video, long timeDiffMillis) {
        Long shareCount = video.getShareCount();
        Double historyCount = video.getHistoryCount() * 0.8;
        Long startCount = video.getStartCount();
        Double favoritesCount = video.getFavoritesCount() * 1.5;
        double random = ThreadLocalRandom.current().nextDouble(100000, 1000000);

        // 避免除零
        long days = TimeUnit.MILLISECONDS.toDays(timeDiffMillis + 1);

        return (shareCount + historyCount + startCount + favoritesCount + random) / days;
    }

    private void finalizeHotRank(String taskId, HotRankEndMessage endMsg) {
        TopK topK = taskTopKMap.remove(taskId);

        if (topK != null) {
            List<HotVideo> hotVideos = topK.get();
            updateRedisHotRank(hotVideos);
            log.info("热点排名任务完成: taskId={}, 总批次数={}, Top10: {}",taskId, endMsg.getTotalBatches(), hotVideos);
        } else {
            log.error("找不到对应的TopK: taskId={}", taskId);
        }
    }

    private void updateRedisHotRank(List<HotVideo> hotVideos) {
        String key = RedisConstant.HOT_RANK;
        redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            for (HotVideo hotVideo : hotVideos) {
                try {
                    hotVideo.setHot(null);
                    // 不这样写铁报错！序列化问题
                    //将string数组转化为bit数组
                    connection.zAdd(key.getBytes(), hotVideo.getHot(), jackson2JsonRedisSerializer.serialize(om.writeValueAsString(hotVideo)));
                } catch (JsonProcessingException e) {
                    e.printStackTrace();
                }
            }
            return null;
        });
    }
}